import java.net.InetAddress
import java.util.UUID

import cn.tecnova.bean.{BaAnalysisBean, BaArticleNlpAnalysisRes, NlpArticleOther}
import cn.tecnova.utils.ESUtils.settings
import cn.tecnova.utils.{BroadcastKafkaProducer, ConfigHandler, ESUtils}
import com.alibaba.fastjson.{JSON, JSONObject}
import com.google.gson.Gson
import org.apache.commons.lang.StringUtils
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.log4j.{Level, Logger}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.elasticsearch.common.settings.Settings
import org.elasticsearch.common.transport.InetSocketTransportAddress
import org.elasticsearch.index.query.QueryBuilders
import org.elasticsearch.search.SearchHit
import org.elasticsearch.transport.client.PreBuiltTransportClient

/**
  * description:
  * Rabcheng
  * Date:2019/6/13 14:19
  **/
object TestBaArticleNlpAnalysis {

  Logger.getLogger("org").setLevel(Level.ERROR)

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf()
      .setAppName(this.getClass.getSimpleName)
      .setMaster("local[*]")

    val sc = new SparkContext(conf)

    val settings = Settings.builder().put("cluster.name", "es-cluster").build()
    val client = new PreBuiltTransportClient(settings)
      //      .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(ConfigHandler.esHost), ConfigHandler.esPort))
      .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("192.168.100.3"), 9300))


    try {

      //用uuid和user_id去es索引库中检索数据
      val builder = client.prepareSearch("ba_user_relation_article").setTypes("ba_user_relation_article")
      val builder1 = QueryBuilders.matchPhraseQuery("template_source", "1")
      val builder2 = QueryBuilders.matchPhraseQuery("uuid", "20190509001")

      builder.setSize(100)
      val sr = builder.setQuery(QueryBuilders.boolQuery()
        .must(builder1)
        .must(builder2))
        .execute()
        .actionGet()
      val hits: Array[SearchHit] = sr.getHits.getHits
      println(hits.length)

      for(hit <- hits){
        println(hit.getSourceAsString)
      }


    } catch {

      case e: Exception => e.printStackTrace()

    } finally {
      //释放资源
      if (client != null) client.close()

    }


    sc.stop()

  }


}
